package com.galudisu.role.front

import java.util.concurrent.atomic.AtomicInteger

import akka.actor.{Actor, ActorLogging, ActorRef, ActorSystem, Props, Terminated}
import akka.pattern._
import akka.util.Timeout
import com.galudisu.comm.{JobFailure, JobSuccess, LogarithmCalcJob, Register}
import com.typesafe.config.ConfigFactory

import scala.concurrent.duration._
import scala.language.postfixOps

/**
  * 角色为Frontend
  *
  * @author galudisu
  */
class ClusterFrontend extends Actor with ActorLogging {

  var backends: IndexedSeq[ActorRef] = IndexedSeq.empty[ActorRef]
  var jobCounter                     = 0

  override def receive: Receive = {

    case job: LogarithmCalcJob if backends.isEmpty =>
      self ! JobFailure("backend nodes is empty!")

    case job: LogarithmCalcJob =>
      jobCounter += 1
      backends(jobCounter % backends.size) forward job

    case Register if !backends.contains(sender()) =>
      context watch sender()
      backends = backends :+ sender()

    case Terminated(a) =>
      backends = backends.filterNot(_ == a)

    case fail: JobFailure => log.info(fail.name)
    case _ => None // do nothing

  }
}

object ClusterFrontend extends App {

  val config   = ConfigFactory.load("frontend")
  val system   = ActorSystem("ClusterSystem", config)
  val frontend = system.actorOf(Props[ClusterFrontend], name = "frontend")

  // 定时发送信息
  val counter = new AtomicInteger

  import system.dispatcher

  system.scheduler.schedule(2.seconds, 2.seconds) {
    implicit val timeout = Timeout(5 seconds)
    (frontend ? LogarithmCalcJob(counter.incrementAndGet())) foreach  {
      case success: JobSuccess => println(success.name)
      case _ => None
    }

  }

}
